﻿using Bit.Art.Core.Extensions;
using Bit.Art.Core.Models;
using Bit.Art.Distributed;
using Disruptor.Dsl;
using Microsoft.Extensions.Logging;
using NetMQ;
using NetMQ.Sockets;
using System;
using System.Collections.Generic;
using System.Text;
using System.Threading.Tasks;
using System.Threading;
using Bit.Art.Core;
using Serilog;

namespace Bit.Art.Engine
{
    public class OrderMatchEngine : IDisposable
    {
        private List<NetMQSocket> _sockets = new List<NetMQSocket>();
        private volatile bool _runing;
        private bool _disposed;
        private PullSocket _puller;
        private PushSocket _bookeeper;
        private RouterSocket _cancelOrderSocket;
        private ResponseSocket _marketChangeSocket;
        private Disruptor<Order> _orderDisruptor;
        private OrderMatchHandler _orderMatchHandler;
        private long _orderno;
        private ReaderWriterLockSlim _globalConfigChangeLock = new ReaderWriterLockSlim();
        // 默认的进入timeout的时间是1秒
        private const int READ_WRITE_LOCK_TIMEOUT_VALUE = 1000;

        //// TODO: disruptor的ring buffer size需要从配置文件里读取
        public Task[] Start(Memcache memcache, int ringBufferSize, int depth, int orderCapacity)
        {
            try
            {
                _runing = true;
                _puller = new PullSocket(string.Format("@{0}", SiteConfiguration.OrderMatchEndpoint));
                _bookeeper = new PushSocket(string.Format(">{0}", SiteConfiguration.MatchResultSinkEndpoint));
                _cancelOrderSocket = new RouterSocket($"@{SiteConfiguration.CancelOrderEndpoint}");
                _marketChangeSocket = new ResponseSocket($"@{SiteConfiguration.ChangeMarketEndpoint}");
                // 将订单放到disruptor的ringbuffer里
                _orderDisruptor = new Disruptor<Order>(() => new Order(), ringBufferSize, TaskScheduler.Default);

                _orderMatchHandler = new OrderMatchHandler(memcache, _bookeeper, _cancelOrderSocket, depth, orderCapacity);
                _orderDisruptor.HandleEventsWith(_orderMatchHandler);
                _orderDisruptor.Start();

                var tasks = new List<Task>();
                tasks.Add(BeginProcessMarketChangeRequests());
                tasks.Add(BeginProcessInComingOrders());
                tasks.Add(BeginProcessCancelOrderRequest());

#if UNITTEST
                Program.s_autoResetEventForTest.Set();
#endif

                return tasks.ToArray();
            }
            catch (Exception e)
            {
#if UNITTEST
                Program.s_autoResetEventForTest.Set();
#endif
                Log.Logger.Error(e, "启动撮合引擎发生错误");
                _runing = false;
                return null;
            }
        }

        private Task BeginProcessMarketChangeRequests()
        {
            return Task.Run(async () =>
            {
                while (_runing)
                {
                    try
                    {
                        var json = _marketChangeSocket.ReceiveFrameString(Encoding.UTF8);
                        byte ret = Constants.ErrorCodes.FAILED;
                        try
                        {
                            var market = Utility.From<Market>(json);
                            var stepRet1 = Constants.ErrorCodes.FAILED;
                            var stepRet2 = stepRet1;
                            try
                            {
                                if (_globalConfigChangeLock.TryEnterWriteLock(READ_WRITE_LOCK_TIMEOUT_VALUE))
                                {
                                    stepRet1 = _orderMatchHandler.UpdateMarket(market);
                                }
                            }
                            finally
                            {
                                _globalConfigChangeLock.ExitWriteLock();
                            }

                            if (stepRet1 == Constants.ErrorCodes.SUCCESS)
                            {
                                using (var client = new RequestSocket(SiteConfiguration.DatabaseEndpoint))
                                {
                                    var msg = new DataMessageBase() { Type = (int)DataMessageBase.MessageType.UpdateMarket };

                                    client.SendFrame(msg.ToJson());
                                    var response = await Task.Run(() =>
                                    {
                                        var res = client.ReceiveMultipartMessage();
                                        var responseJson = res.Last.ConvertToString(System.Text.Encoding.UTF8);
                                        return Utility.From<ApiCallResult>(responseJson);
                                    });
                                    stepRet2 = response.Success ? Constants.ErrorCodes.SUCCESS : Constants.ErrorCodes.FAILED;
                                }
                            }

                            ret = stepRet1 == Constants.ErrorCodes.SUCCESS && stepRet2 == Constants.ErrorCodes.SUCCESS ?
                                                       Constants.ErrorCodes.SUCCESS : Constants.ErrorCodes.FAILED;
                        }
                        finally
                        {
                            _marketChangeSocket.SendFrame(new byte[] { ret });
                        }
                    }
                    catch (Exception ex)
                    {
                        Log.Logger.Error(ex, "修改交易对配置时出错！");
                    }
                }
            });
        }


        private Task BeginProcessCancelOrderRequest()
        {
            return Task.Run(() =>
            {
                var dataCopier = new OrderEventTranslator();
                while (_runing)
                {
                    try
                    {
                        var msg = _cancelOrderSocket.ReceiveMultipartMessage();
                        var order = Order.FromBytes(msg.Last.Buffer);

                        try
                        {
                            if (_globalConfigChangeLock.TryEnterReadLock(READ_WRITE_LOCK_TIMEOUT_VALUE))
                            {
                                _orderDisruptor.PublishEvent(dataCopier, order);
                            }
                            else
                            {
                                // TODO: 需要增加万一Market获取不到读写锁情况下的处理代码
                                Log.Logger.Warning($"由于无法获取全局配置读写锁的原因，取消订单{order.Id}失败！");
                            }
                        }
                        finally
                        {
#if UNITTEST
                            Program.s_autoResetEventForTest.Set();
#endif
                            _globalConfigChangeLock.ExitReadLock();
                        }
                    }
                    catch (Exception ex)
                    {
                        Log.Logger.Error(ex, "撮合引擎中取消订单模块发生严重错误");
                    }
                }
            });
        }

        private Task BeginProcessInComingOrders()
        {
            return Task.Run(() =>
            {
                var dataCopier = new OrderEventTranslator();
                while (_runing)
                {
                    try
                    {
                        var msg = _puller.ReceiveMultipartMessage();
                        var order = Order.FromBytes(msg.Last.Buffer);
                        Interlocked.Increment(ref _orderno);
                        Log.Logger.Debug($"[Engine] got order no: {_orderno}, side: {order.Side}, price {order.Price}, volume {order.Volume}");

                        try
                        {
                            if (_globalConfigChangeLock.TryEnterReadLock(READ_WRITE_LOCK_TIMEOUT_VALUE))
                            {
                                _orderDisruptor.PublishEvent(dataCopier, order);
                            }
                            else
                            {
                                // TODO: 需要增加万一Market获取不到读写锁情况下的处理代码
                                // TODO: 需要检查加了这个读写锁的性能问题
                                Log.Logger.Warning($"由于无法获取全局配置读写锁的原因，添加订单{order.Id}失败，失败的订单详情：\n{order.ToJson()}");
                            }
                        }
                        finally
                        {
                            _globalConfigChangeLock.ExitReadLock();
                        }
                    }
                    catch (Exception ex)
                    {
                        Log.Logger.Error(ex, "撮合引擎循环碰到严重错误");
                    }
                }
            });
        }

        public bool Stop()
        {
            try
            {
                _runing = false;
                if (_puller != null)
                {
                    _puller.Dispose();
                    _puller = null;
                }
                if (_bookeeper != null)
                {
                    _bookeeper.Dispose();
                    _bookeeper = null;
                }
                if (_cancelOrderSocket != null)
                {
                    _cancelOrderSocket.Dispose();
                    _cancelOrderSocket = null;
                }
                _orderDisruptor.Halt();
                return true;
            }
            catch (Exception e)
            {
                Log.Logger.Error(e, "启动撮合引擎发生错误");
                _runing = false;
                return false;
            }
        }

        public void Dispose()
        {
            if (!_disposed)
            {
                Stop();
                _disposed = true;
            }
        }
    }
}
